#ifndef KAFKA_H_
#define KAFKA_H_

#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <sys/stat.h>
#include <netdb.h>
#include <errno.h>
#include <arpa/inet.h>
#include <sys/stat.h>
#include <string>
#include <list>
#include <assert.h>
#include <string.h>
#include <iostream>
#include <librdkafka/rdkafkacpp.h>

struct kafka_parse
{
  std::string kafka_Ip;
  std::string kafka_Port; //默认端口21
  std::string kafka_SavePath;
  std::string kafka_UserName;
  std::string kafka_Password;
};

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb
{
public:
  void dr_cb(RdKafka::Message &message)
  {
    /* If message.err() is non-zero the message delivery failed permanently
     * for the message. */
    if (message.err())
      std::cerr << "% Message delivery failed: " << message.errstr() << std::endl;
    else
      std::cerr << "% Message delivered to topic " << message.topic_name() << " [" << message.partition() << "] at offset " << message.offset() << std::endl;
  }
};

class KAFKAManager
{
public:
  ~KAFKAManager();
  void createProducer(const std::string &brokers);
  void produceMessage(const std::string &topic, char *data, const int &data_len, char *key, const int &key_len);
  RdKafka::Producer *producer;
  ExampleDeliveryReportCb ex_dr_cb;
};

#endif